package com.ruicar.afs.cloud.common.mq.rabbit.listener;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.rabbitmq.client.Channel;
import com.ruicar.afs.cloud.common.core.enums.AfsEnumUtil;
import com.ruicar.afs.cloud.common.core.service.ServiceInfo;
import com.ruicar.afs.cloud.common.mq.rabbit.helper.CacheInfo;
import com.ruicar.afs.cloud.common.mq.rabbit.helper.bean.ServerQueueRef;
import com.ruicar.afs.cloud.common.mq.rabbit.message.AfsTransEntity;
import com.ruicar.afs.cloud.common.mq.rabbit.message.RabbitMessageWrapper;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.util.Assert;

import java.io.IOException;
import java.lang.reflect.Method;
import java.text.MessageFormat;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * The type AfsAbstractListener
 * <p>Description: </p>
 *
 * @author Fzero
 * @version 1.0
 * @date Created on 2020.06.11 21:25:17
 * @since 1.0
 */
@Slf4j
public abstract class AfsAbstractListener {
    private static final Logger LISTENER_LOG = LoggerFactory.getLogger("mq-message-listener");
    /**
     * Message receive format
     * <p>Description: </p>
     * 0 queueName
     * 1 serviceName
     * 2 messageId
     */
    private final String MESSAGE_RECEIVE_FORMAT = "afs_rabbit_receive:{0}:{1}:{2}";


    /**
     * Redis key ttl time
     * <p>Description: </p>
     * redis key 存在周期120分钟
     */
    private final long REDIS_KEY_TTL_TIME = 120 * 60;
    /**
     * String redis template
     * <p>Description: </p>
     */
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    private static final Map<Class, Class> LISTENER_CACHE = new ConcurrentHashMap<>(8);

    /**
     * Process *
     *
     * @param channel channel
     * @param message message
     */
    protected void process(Channel channel, Message message) {
        logMessage(message.getMessageProperties().getConsumerQueue(), new String(message.getBody()));
        RabbitMessageWrapper rabbitMessageWrapper = getWrapperMessage(message.getBody());
        try {
            if (rabbitMessageWrapper != null && rabbitMessageWrapper.isAfsMessage()) {
                processAfsPlatformMessage(channel, message, rabbitMessageWrapper);
            } else {
                log.info("消息未来自afs,无法确认消息id，不做幂等检测");
                processAfsMethodMessage(channel, message, new String(message.getBody()), false, null);
            }
        } catch (Throwable e) {
            log.error("", e);
        }
    }

    private Object makeObject(RabbitMessageWrapper rabbitMessageWrapper, Class<? extends AfsMqBizProcessor> clazz) throws Exception {
        if (!LISTENER_CACHE.containsKey(clazz)) {
            for (Method method : clazz.getMethods()) {
                if (method.getName().equals("processMessage") && method.getParameterCount() == 1 && !(Object.class.equals(method.getParameterTypes()[0]))) {
                    LISTENER_CACHE.put(clazz, method.getParameterTypes()[0]);
                    break;
                }
            }
            Assert.isTrue(LISTENER_CACHE.containsKey(clazz), clazz.getName() + "处理类定义不正确，processMessage方法必须接收非自定义对象");
        }
        return parseResult(rabbitMessageWrapper.getContent(), LISTENER_CACHE.get(clazz)).getData();
    }

    private boolean processAfsMethodMessage(Channel channel, Message message, String content, boolean checkFlag, String id) {
        try {
            try {
                ServerQueueRef serverQueueRef = CacheInfo.CACHE_SERVER_QUEUE.get(getCacheKey());
                if (serverQueueRef.isTransEntity()) {
                    serverQueueRef.getMethod().invoke(serverQueueRef.getObject(), parseResult(content, serverQueueRef.getTransBodyType()));
                } else {
                    if (serverQueueRef.getTargetClass().equals(String.class)) {
                        serverQueueRef.getMethod().invoke(serverQueueRef.getObject(), content);
                    } else {
                        serverQueueRef.getMethod().invoke(serverQueueRef.getObject(), JSONObject.parseObject(content, serverQueueRef.getTargetClass()));
                    }
                }
                if (checkFlag) {
                    stringRedisTemplate.opsForValue().set(MessageFormat.format(MESSAGE_RECEIVE_FORMAT, getQueueName(), ServiceInfo.getServiceName(), id), "consumer", REDIS_KEY_TTL_TIME, TimeUnit.SECONDS);
                }
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Throwable e) {
                log.error("rabbit 消息接收失败", e);
                if (CacheInfo.CACHE_SERVER_QUEUE.get(getCacheKey()).isRequeue()) {
                    log.info("消息接收处理失败，消息tag{},服务器别名{},队列名称{},配置是否重新队列{}，报文重新队列", message.getMessageProperties().getDeliveryTag(), getServerAlias(), getQueueName(), CacheInfo.CACHE_SERVER_QUEUE.get(getCacheKey()).isRequeue());
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                } else {
                    log.info("消息接收处理失败，消息tag{},服务器别名{},队列名称{},配置是否重新队列{}，丢弃报文:{}", message.getMessageProperties().getDeliveryTag(), getServerAlias(), getQueueName(), CacheInfo.CACHE_SERVER_QUEUE.get(getCacheKey()).isRequeue(), new String(message.getBody()));
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                }
            }
        } catch (Throwable t) {
            try {
                //fix 防止从控制台发送不规范报文导致产生uAck消息，导致队列堵塞
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }catch (IOException ioE){
                log.error("异常消息确认失败",ioE);
            }
            log.error("处理消息异常,做消息确认处理", t);
        }
        return true;
    }

    private void processAfsPlatformMessage(Channel channel, Message message, RabbitMessageWrapper rabbitMessageWrapper) {
        try {
            if (stringRedisTemplate.opsForValue().get(MessageFormat.format(MESSAGE_RECEIVE_FORMAT, getQueueName(), ServiceInfo.getServiceName(), rabbitMessageWrapper.getAfsMessageId())) == null) {
                log.info("消息{}未消费", rabbitMessageWrapper.getAfsMessageId());
                //消体为  afsMessageEntity
                if (rabbitMessageWrapper.isAfsEntity()) {
                    try {
                        if (!CacheInfo.CACHE_TRANS_CODE_PROCESS.containsKey(rabbitMessageWrapper.getTransCode())) {
                            log.warn("交易码{}未定义处理器，丢弃报文{}", AfsEnumUtil.key(rabbitMessageWrapper.getTransCode()), rabbitMessageWrapper.getContent());
                            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                            return;
                        }
                        AfsMqBizProcessor afsMqBizProcessor = CacheInfo.CACHE_TRANS_CODE_PROCESS.get(rabbitMessageWrapper.getTransCode());
                        boolean result = afsMqBizProcessor.processMessage(makeObject(rabbitMessageWrapper, afsMqBizProcessor.getClass()));
                        if (result) {
                            stringRedisTemplate.opsForValue().set(MessageFormat.format(MESSAGE_RECEIVE_FORMAT, getQueueName(), ServiceInfo.getServiceName(), rabbitMessageWrapper.getAfsMessageId()), "consumer", REDIS_KEY_TTL_TIME, TimeUnit.SECONDS);
                        }
                        if (!result && afsMqBizProcessor.reQueueOnFail()) {
                            log.warn("交易码[{}]--消息ID===>[{}]--消息处理失败，消息返回队列", AfsEnumUtil.key(rabbitMessageWrapper.getTransCode()), rabbitMessageWrapper.getAfsMessageId());
                            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                        } else {
                            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                        }
                    } catch (Throwable t) {
                        log.error("rabbit 消息接收失败", t);
                        if (CacheInfo.CACHE_TRANS_CODE_PROCESS.get(rabbitMessageWrapper.getTransCode()).reQueueOnFail()) {
                            log.info("消息接收处理失败，消息tag{},服务器别名{},队列名称{},配置是否重新队列{}，报文重新队列", message.getMessageProperties().getDeliveryTag(), getServerAlias(), getQueueName(), CacheInfo.CACHE_TRANS_CODE_PROCESS.get(rabbitMessageWrapper.getTransCode()).reQueueOnFail());
                            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                        } else {
                            log.info("消息接收处理失败，消息tag{},服务器别名{},队列名称{},配置是否重新队列{}，丢弃报文:{}", message.getMessageProperties().getDeliveryTag(), getServerAlias(), getQueueName(), CacheInfo.CACHE_TRANS_CODE_PROCESS.get(rabbitMessageWrapper.getTransCode()).reQueueOnFail(), new String(message.getBody()));
                            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                        }
                    }
                } else {
                    processAfsMethodMessage(channel, message, rabbitMessageWrapper.getContent(), true, rabbitMessageWrapper.getAfsMessageId());
                }
            } else {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }
        } catch (Throwable e) {
            //fix 防止从控制台发送不规范报文导致产生uAck消息，导致队列堵塞
            try {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }catch (IOException ioE){
                log.error("异常消息确认失败",ioE);
            }
            log.error("处理消息异常,做消息确认处理", e);
        }
    }


    private static <T> AfsTransEntity<T> parseResult(String json, Class<T> clazz) {
        return JSONObject.parseObject(json, new TypeReference<AfsTransEntity<T>>(clazz) {
        });
    }

    private RabbitMessageWrapper getWrapperMessage(byte[] message) {
        try {
            return JSONObject.parseObject(message, RabbitMessageWrapper.class);
        } catch (Exception e) {
            return null;
        }
    }

    private String getCacheKey() {
        return getServerAlias() + "-" + getQueueName();
    }

    public abstract String getServerAlias();

    public abstract String getQueueName();

    private void logMessage(String name, String content) {
        LISTENER_LOG.info("接收到mq报文，队列名:[{}],报文体\n:{}", name, content);
    }

    public abstract void listener(Channel channel, Message message);
}
